package com.youkeda.kafka.study.config;


import com.alibaba.fastjson.JSON;
import com.youkeda.kafka.study.model.Student;
import com.youkeda.kafka.study.model.SummaryData;
import com.youkeda.kafka.study.service.StudyDataService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.support.serializer.JsonSerde;


@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfiguration {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsConfiguration.class);

    @Autowired
    private StudyDataService studyDataService;

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("studyData");
        stream.map((key, value)->{

            Student student;

             try {
                 student= JSON.parseObject(value,Student.class);
//                 System.out.println(student.getName());
             }catch (Exception e){
                 LOG.error("receive error data. key=" + key + ", value=" + value);
                 student = new Student();
             }
            return new KeyValue<>(key, student);
        }).filter((key,s)->s.getName()!=null&&s.getStdudyDate()!=null).foreach((key,s)->{
            SummaryData sd = studyDataService.findStudyData(s.getName());
            if (null==sd) {
                sd = new SummaryData();
                sd.setName(s.getName());
                sd.setDateCount(1);
                sd.setTotalCodeRows(s.getCodeRows());
                sd.setTotalSections(s.getSections());
                sd.setTotalTime(s.getDuration());
                studyDataService.saveStudyData(sd);
            }else{
                sd.setDateCount(sd.getDateCount() + 1);
                sd.setTotalTime(sd.getTotalTime() + s.getDuration());
                sd.setTotalSections(sd.getTotalSections() + s.getSections());
                sd.setTotalCodeRows(sd.getTotalCodeRows() + s.getCodeRows());
                studyDataService.saveStudyData(sd);
            }

        });


        KStream<String,Student> studentKStream = stream.map((key,value)->{
            Student student;

            try {
                student= JSON.parseObject(value,Student.class);
//                System.out.println(student.getName());
            }catch (Exception e){
                LOG.error("receive error data. key=" + key + ", value=" + value);
                student = new Student();
            }
            return new KeyValue<>(key, student);
        }).filter((key,student)->{
            return StringUtils.isNotBlank(student.getName())&& null!=student.getStdudyDate();
        });

        // 学习天数
        studentKStream.to("studyDateCount", Produced.with(Serdes.String(), new JsonSerde<>()));
        // 学习总时长
        studentKStream.to("studyDuration", Produced.with(Serdes.String(), new JsonSerde<>()));
        // 学习节数
        studentKStream.to("studySections", Produced.with(Serdes.String(), new JsonSerde<>()));
        // 代码行数
        studentKStream.to("studyCodeRows", Produced.with(Serdes.String(), new JsonSerde<>()));

        return stream;
    }


}
